在 schedule 函数中，我们简单提过找一个 runnable goroutine 的过程，这一讲我们来详细分析源码。

工作线程 M 费尽心机也要找到一个可运行的 goroutine，这是它的工作和职责，不达目的，绝不罢体，这种锲而不舍的精神值得每个人学习。

共经历三个过程：先从本地队列找，定期会从全局队列找，最后实在没办法，就去别的 P 偷。如下图所示：

M 找工作的过程

先看第一个：从 P 本地队列找。源码如下：

  1. // 从本地可运行队列里找到一个 g
  2. // 如果 inheritTime 为真，gp 应该继承这个时间片，否则，新开启一个时间片
  3. func runqget(_p_ *p) (gp *g, inheritTime bool) {
  4.     // If there's a runnext, it's the next G to run.
  5.     // 如果 runnext 不为空，则 runnext 是下一个待运行的 G
  6.     for {
  7.         next := _p_.runnext
  8.         if next == 0 {
  9.             // 为空，则直接跳出循环
  10.             break
  11.         }
  12.         // 再次比较 next 是否没有变化
  13.         if _p_.runnext.cas(next, 0) {
  14.             // 如果没有变化，则返回 next 所指向的 g。且需要继承时间片
  15.             return next.ptr(), true
  16.         }
  17.     }
  19.     for {
  20.         // 获取队列头
  21.         h := atomic.Load(&_p_.runqhead) // load-acquire, synchronize with other consumers
  22.         // 获取队列尾
  23.         t := _p_.runqtail
  24.         if t == h {
  25.             // 头和尾相等，说明本地队列为空，找不到 g
  26.             return nil, false
  27.         }
  28.         // 获取队列头的 g
  29.         gp := _p_.runq[h%uint32(len(_p_.runq))].ptr()
  30.         // 原子操作，防止这中间被其他线程因为偷工作而修改
  31.         if atomic.Cas(&_p_.runqhead, h, h+1) { // cas-release, commits consume
  32.             return gp, false
  33.         }
  34.     }
  35. }

整个源码结构比较简单，主要是两个 for 循环。

第一个 for 循环尝试返回 P 的 runnext 成员，因为 runnext 具有最高的运行优先级，因此要首先尝试获取 runnext。当发现 runnext 为空时，直接跳出循环，进入第二个。否则，用原子操作获取 runnext，并将其值修改为 0，也就是空。这里用到原子操作的原因是防止在这个过程中，有其他线程过来“偷工作”，导致并发修改 runnext 成员。

第二个 for 循环则是在尝试获取 runnext 成员失败后，尝试从本地队列中返回队列头的 goroutine。同样，先用原子操作获取队列头，使用原子操作的原因同样是防止其他线程“偷工作”时并发对队列头的并发写操作。之后，直接获取队列尾，因为不担心其他线程同时更改，所以直接获取。注意，“偷工作”时只会修改队列头。

比较队列头和队列尾，如果两者相等，说明 P 本地队列没有可运行的 goroutine，直接返回空。否则，算出队列头指向的 goroutine，再用一个 CAS 原子操作来尝试修改队列头，使用原子操作的原因同上。

从本地队列获取可运行 goroutine 的过程比较简单，我们再来看从全局队列获取 goroutine 的过程。在 schedule 函数中调用 globrunqget 的代码：

  1. // 为了公平，每调用 schedule 函数 61 次就要从全局可运行 goroutine 队列中获取
  2. if _g_.m.p.ptr().schedtick%61 == 0 && sched.runqsize > 0 {
  3.     lock(&sched.lock)
  4.     // 从全局队列最大获取 1 个 gorutine
  5.     gp = globrunqget(_g_.m.p.ptr(), 1)
  6.     unlock(&sched.lock)
  7. }

这说明，并不是每次调度都会从全局队列获取可运行的 goroutine。实际情況是调度器每调度 61 次并且全局队列有可运行 goroutine 的情况下才会调用 globrunqget 函数尝试从全局获取可运行 goroutine。毕竟，从全局获取需要上锁，这个开销可就大了，能不做就不做。

我们来详细看下 globrunqget 的源码：

  1. // 尝试从全局队列里获取可运行的 goroutine 队列
  2. func globrunqget(_p_ *p, max int32) *g {
  3.     // 如果队列大小为 0
  4.     if sched.runqsize == 0 {
  5.         return nil
  6.     }
  8.     // 根据 p 的数量平分全局运行队列中的 goroutines
  9.     n := sched.runqsize/gomaxprocs + 1
  10.     if n > sched.runqsize {
  11.         n = sched.runqsize // 如果 gomaxprocs 为 1
  12.     }
  14.     // 修正"偷"的数量
  15.     if max > 0 && n > max {
  16.         n = max
  17.     }
  18.     // 最多只能"偷"本地工作队列一半的数量
  19.     if n > int32(len(_p_.runq))/2 {
  20.         n = int32(len(_p_.runq)) / 2
  21.     }
  23.     // 更新全局可运行队列长度
  24.     sched.runqsize -= n
  25.     // 如果都要被"偷"走，修改队列尾
  26.     if sched.runqsize == 0 {
  27.         sched.runqtail = 0
  28.     }
  30.     // 获取队列头指向的 goroutine
  31.     gp := sched.runqhead.ptr()
  32.     // 移动队列头
  33.     sched.runqhead = gp.schedlink
  34.     n--
  35.     for ; n > 0; n-- {
  36.         // 获取当前队列头
  37.         gp1 := sched.runqhead.ptr()
  38.         // 移动队列头
  39.         sched.runqhead = gp1.schedlink
  40.         // 尝试将 gp1 放入 P 本地，使全局队列得到更多的执行机会
  41.         runqput(_p_, gp1, false)
  42.     }
  43.     // 返回最开始获取到的队列头所指向的 goroutine
  44.     return gp
  45. }

代码比较简单。首先根据全局队列的可运行 goroutine 长度和 P 的总数，来计算一个数值，表示每个 P 可平均分到的 goroutine 数量。

然后根据函数参数中的 max 以及 P 本地队列的长度来决定把多少全局队列中的 goroutine 转移到 P 本地。

最后，for 循环挨个把全局队列中 n-1 个 goroutine 转移到本地，并且返回最开始获取到的队列头所指向的 goroutine，毕竟它最需要得到运行的机会。

把全局队列中的可运行 goroutine 转移到本地队列，给了全局队列中可运行 goroutine 运行的机会，不然全局队列中的 goroutine 一直得不到运行。

最后，我们继续看第三个过程，从其他 P “偷工作”：

  1. // 从本地运行队列和全局运行队列都没有找到需要运行的 goroutine，
  2. // 调用 findrunnable 函数从其它工作线程的运行队列中偷取，如果偷不到，则当前工作线程进入睡眠
  3. // 直到获取到 runnable goroutine 之后 findrunnable 函数才会返回。
  4. if gp == nil {
  5.     gp, inheritTime = findrunnable() // blocks until work is available
  6. }

这是整个找工作过程最复杂的部分：

  1. // 从其他地方找 goroutine 来执行
  2. func findrunnable() (gp *g, inheritTime bool) {
  3.     _g_ := getg()
  5. top:
  6.     _p_ := _g_.m.p.ptr()
  8.     // ……………………
  10.     // local runq
  11.     // 从本地队列获取
  12.     if gp, inheritTime := runqget(_p_); gp != nil {
  13.         return gp, inheritTime
  14.     }
  16.     // global runq
  17.     // 从全局队列获取
  18.     if sched.runqsize != 0 {
  19.         lock(&sched.lock)
  20.         gp := globrunqget(_p_, 0)
  21.         unlock(&sched.lock)
  22.         if gp != nil {
  23.             return gp, false
  24.         }
  25.     }
  27.     // ……………………
  29.     // Steal work from other P's.
  31.     // 如果其他的 P 都处于空闲状态，那肯定没有其他工作要做
  32.     procs := uint32(gomaxprocs)
  33.     if atomic.Load(&sched.npidle) == procs-1 {
  34.         goto stop
  35.     }
  37.     // 如果有很多工作线程在找工作，那我就停下休息。避免消耗太多 CPU
  38.     if !_g_.m.spinning && 2*atomic.Load(&sched.nmspinning) >= procs-atomic.Load(&sched.npidle) {
  39.         goto stop
  40.     }
  42.     if !_g_.m.spinning {
  43.         // 设置自旋状态为 true
  44.         _g_.m.spinning = true
  45.         // 自旋状态数加 1
  46.         atomic.Xadd(&sched.nmspinning, 1)
  47.     }
  48.     // 从其它 p 的本地运行队列盗取 goroutine
  49.     for i := 0; i < 4; i++ {
  50.         for enum := stealOrder.start(fastrand()); !enum.done(); enum.next() {
  51.             // ……………………
  52.             stealRunNextG := i > 2 // first look for ready queues with more than 1 g
  53.             if gp := runqsteal(_p_, allp[enum.position()], stealRunNextG); gp != nil {
  54.                 return gp, false
  55.             }
  56.         }
  57.     }
  59. stop:
  61.     // ……………………
  63.     // return P and block
  64.     lock(&sched.lock)
  65.     if sched.gcwaiting != 0 || _p_.runSafePointFn != 0 {
  66.         unlock(&sched.lock)
  67.         goto top
  68.     }
  69.     if sched.runqsize != 0 {
  70.         gp := globrunqget(_p_, 0)
  71.         unlock(&sched.lock)
  72.         return gp, false
  73.     }
  74.     // 当前工作线程解除与 p 之间的绑定，准备去休眠
  75.     if releasep() != _p_ {
  76.         throw("findrunnable: wrong p")
  77.     }
  78.     // 把 p 放入空闲队列
  79.     pidleput(_p_)
  80.     unlock(&sched.lock)
  82.     wasSpinning := _g_.m.spinning
  83.     if _g_.m.spinning {
  84.         // m 即将睡眠，不再处于自旋
  85.         _g_.m.spinning = false
  86.         if int32(atomic.Xadd(&sched.nmspinning, -1)) < 0 {
  87.             throw("findrunnable: negative nmspinning")
  88.         }
  89.     }
  91.     // check all runqueues once again
  92.     // 休眠之前再检查一下所有的 p，看一下是否有工作要做
  93.     for i := 0; i < int(gomaxprocs); i++ {
  94.         _p_ := allp[i]
  95.         if _p_ != nil && !runqempty(_p_) {
  96.             lock(&sched.lock)
  97.             _p_ = pidleget()
  98.             unlock(&sched.lock)
  99.             if _p_ != nil {
  100.                 acquirep(_p_)
  101.                 if wasSpinning {
  102.                     _g_.m.spinning = true
  103.                     atomic.Xadd(&sched.nmspinning, 1)
  104.                 }
  105.                 goto top
  106.             }
  107.             break
  108.         }
  109.     }
  111.     // ……………………
  113.     // 休眠
  114.     stopm()
  115.     goto top
  116. }

这部分也是最能说明 M 找工作的锲而不舍精神：尽力去各个运行队列中寻找 goroutine，如果实在找不到则进入睡眠状态，等待有工作时，被其他 M 唤醒。

先获取当前指向的 g，也就是 g0，然后拿到其绑定的 p，即 _p_。

首先再次尝试从 _p_ 本地队列获取 goroutine，如果没有获取到，则尝试从全局队列获取。如果还没有获取到就会尝试去“偷”了，这也是没有办法的事。

不过，在偷之前，先看大的局势。如果其他所有的 P 都处于空闲状态，就说明其他 P 肯定没有工作可做，就没必要再去偷了，毕竟“地主家也没有余粮了”，跳到 stop 部分。接着再看下当前正在“偷工作”的线程数量“太多了”，就没必要扎堆了，这么多人，竞争肯定大，工作肯定不好找，也不好偷。

在真正的“偷”工作之前，把自己的自旋状态设置为 true，全局自旋数量加 1。

终于到了“偷工作”的部分了，好紧张！整个过程由两层 for 循环组成，外层控制尝试偷的次数，内层控制“偷”的顺序，并真正的去“偷”。实际上，内层会遍历所有的 P，因此，整体看来，会尝试 4 次扫遍所有的 P，并去“偷工作”，是不是非常有毅力！

第二层的循环并不是每次都按一个固定的顺序去遍历所有的 P，这样不太科学，而是使用了一些方法，“随机”地遍历。具体是使用了下面这个变量：

  1. var stealOrder randomOrder
  3. type randomOrder struct {
  4.     count    uint32
  5.     coprimes []uint32
  6. }

初始化的时候会给 count 赋一个值，例如 8，根据 count 计算出 coprimes，里面的元素是小于 count 的值，且和 8 互质，算出来是：[1, 3, 5, 7]。

第二层循环，开始随机给一个值，例如 2，则第一个访问的 P 就是 P2；从 coprimes 里取出索引为 2 的值为 5，那么，第二个访问的 P 索引就是 2+5=7；依此类推，第三个就是 7+5=12，和 count 做一个取余操作，即 12%8=4……

在最后一次遍历所有的 P 的过程中，连人家的 runnext 也要尝试偷过来，毕竟前三次的失败经验证明，工作太不好“偷”了，民不聊生啊，只能做得绝一点了，stealRunNextG 控制是否要打 runnext 的主意：

  1. stealRunNextG := i > 2

确定好准备偷的对象 allp[enum.position() 之后，调用 runqsteal(_p_, allp[enum.position()], stealRunNextG) 函数执行。

  1. // 从 p2 偷走一半的工作放到 _p_ 的本地
  2. func runqsteal(_p_, p2 *p, stealRunNextG bool) *g {
  3.     // 队尾
  4.     t := _p_.runqtail
  5.     // 从 p2 偷取工作，放到 _p_.runq 的队尾
  6.     n := runqgrab(p2, &_p_.runq, t, stealRunNextG)
  7.     if n == 0 {
  8.         return nil
  9.     }
  10.     n--
  11.     // 找到最后一个 g，准备返回
  12.     gp := _p_.runq[(t+n)%uint32(len(_p_.runq))].ptr()
  13.     if n == 0 {
  14.         // 说明只偷了一个 g
  15.         return gp
  16.     }
  17.     // 队列头
  18.     h := atomic.Load(&_p_.runqhead) // load-acquire, synchronize with consumers
  19.     // 判断是否偷太多了
  20.     if t-h+n >= uint32(len(_p_.runq)) {
  21.         throw("runqsteal: runq overflow")
  22.     }
  23.     // 更新队尾，将偷来的工作加入队列
  24.     atomic.Store(&_p_.runqtail, t+n) // store-release, makes the item available for consumption
  25.     return gp
  26. }

调用 runqgrab 从 p2 偷走它一半的工作放到 _p_ 本地：

  1. n := runqgrab(p2, &_p_.runq, t, stealRunNextG)

runqgrab 函数将从 p2 偷来的工作放到以 t 为地址的数组里，数组就是 _p_.runq。 我们知道，t 是 _p_.runq 的队尾，因此这行代码表达的真正意思是将从 p2 偷来的工作，神不知，鬼不觉地放到 _p_.runq 的队尾，之后，再悄悄改一下 `_p_.runqtail 就把这些偷来的工作据为己有了。

接着往下看，返回的 n 表示偷到的工作数量。先将 n 自减 1，目的是把第 n 个工作（也就是 g）直接返回，如果这时候 n 变成 0 了，说明就只偷到了一个 g，那就直接返回。否则，将队尾往后移动 n，把偷来的工作合法化，简直完美！

我们接着往下看 runqgrab 函数的实现：

  1. // 从 _p_ 批量获取可运行 goroutine，放到 batch 数组里
  2. // batch 是一个环，起始于 batchHead
  3. // 返回偷的数量，返回的 goroutine 可被任何 P 执行
  4. func runqgrab(_p_ *p, batch *[256]guintptr, batchHead uint32, stealRunNextG bool) uint32 {
  5.     for {
  6.         // 队列头
  7.         h := atomic.Load(&_p_.runqhead) // load-acquire, synchronize with other consumers
  8.         // 队列尾
  9.         t := atomic.Load(&_p_.runqtail) // load-acquire, synchronize with the producer
  10.         // g 的数量
  11.         n := t - h
  12.         // 取一半
  13.         n = n - n/2
  14.         if n == 0 {
  15.             if stealRunNextG {
  16.                 // 连 runnext 都要偷，没有人性
  17.                 // Try to steal from _p_.runnext.
  18.                 if next := _p_.runnext; next != 0 {
  19.                     // 这里是为了防止 _p_ 执行当前 g，并且马上就要阻塞，所以会马上执行 runnext，
  20.                     // 这个时候偷就没必要了，因为让 g 在 P 之间"游走"不太划算，
  21.                     // 就不偷了，给他们一个机会。
  22.                     // channel 一次同步的的接收发送需要 50ns 左右，因此 3us 差不多给了他们 50 次机会了，做得还是不错的
  23.                     if GOOS != "windows" {
  24.                         usleep(3)
  25.                     } else {
  26.                         osyield()
  27.                     }
  28.                     if !_p_.runnext.cas(next, 0) {
  29.                         continue
  30.                     }
  31.                     // 真的偷走了 next
  32.                     batch[batchHead%uint32(len(batch))] = next
  33.                     // 返回偷的数量，只有 1 个
  34.                     return 1
  35.                 }
  36.             }
  37.             // 没偷到
  38.             return 0
  39.         }
  40.         // 如果 n 这时变得太大了，重新来一遍了，不能偷的太多，做得太过分了
  41.         if n > uint32(len(_p_.runq)/2) { // read inconsistent h and t
  42.             continue
  43.         }
  44.         // 将 g 放置到 bacth 中
  45.         for i := uint32(0); i < n; i++ {
  46.             g := _p_.runq[(h+i)%uint32(len(_p_.runq))]
  47.             batch[(batchHead+i)%uint32(len(batch))] = g
  48.         }
  49.         // 工作被偷走了，更新一下队列头指针
  50.         if atomic.Cas(&_p_.runqhead, h, h+n) { // cas-release, commits consume
  51.             return n
  52.         }
  53.     }
  54. }

外层直接就是一个无限循环，先用原子操作取出 p 的队列头和队列尾，算出一半的 g 的数量，如果 n == 0，说明地主家也没有余粮，这时看 stealRunNextG 的值。如果为假，说明不偷 runnext，那就直接返回 0，啥也没偷到；如果为真，则要尝试偷一下 runnext。

先判断 runnext 不为空，那就真的准备偷了。不过在这之前，要先休眠 3 us。这是为了防止 p 正在执行当前的 g，马上就要阻塞（可能是向一个非缓冲的 channel 发送数据，没有接收者），之后会马上执行 runnext。这个时候偷就没必要了，因为 runnext 马上就要执行了，偷走它还不是要去执行，那何必要偷呢？大家的愿望就是提高效率，这样让 g 在 P 之间”游走”不太划算，索性先不偷了，给他们一个机会。channel 一次同步的的接收或发送需要 50ns 左右，因此休眠 3us 差不多给了他们 50 次机会了，做得还是挺厚道的。

继续看，再次判断 n 是否小于等于 p.runq 长度的一半，因为这个时候很可能 p 也被其他线程偷了，它的 p.runq 就没那么多工作了，这个时候就不能偷这么多了，要重新再走一次循环。

最后一个 for 循环，将 p.runq 里的 g 放到 batch 数组里。使用原子操作更新 p 的队列头指针，往后移动 n 个位置，这些都是被偷走的，伤心！

回到 findrunnable 函数，经过上述三个层面的“偷窃”过程，我们仍然没有找到工作，真惨！于是就走到了 stop 这个代码块。

先上锁，因为要将 P 放到全局空闲 P 链表里去。在这之前还不死心，再瞧一下全局队列里是否有工作，如果有，再去尝试偷全局。

如果没有，就先解除当前工作线程和当前 P 的绑定关系：

  1. // 解除 p 与 m 的关联
  2. func releasep() *p {
  3.     _g_ := getg()
  5.     // ……………………
  7.     _p_ := _g_.m.p.ptr()
  9.     // ……………………
  11.     // 清空一些字段
  12.     _g_.m.p = 0
  13.     _g_.m.mcache = nil
  14.     _p_.m = 0
  15.     _p_.status = _Pidle
  16.     return _p_
  17. }

主要的工作就是将 p 的 m 字段清空，并将 p 的状态修改为 _Pidle。

这之后，将其放入全局空闲 P 列表：

  1. // 将 p 放到 _Pidle 列表里
  2. //go:nowritebarrierrec
  3. func pidleput(_p_ *p) {
  4.     if !runqempty(_p_) {
  5.         throw("pidleput: P has non-empty run queue")
  6.     }
  7.     _p_.link = sched.pidle
  8.     sched.pidle.set(_p_)
  9.     // 增加全局空闲 P 的数量
  10.     atomic.Xadd(&sched.npidle, 1) // TODO: fast atomic
  11. }

构造链表的过程其实比较简单，先将 p.link 指向原来的 sched.pidle 所指向的 p，也就是原空闲链表的最后一个 P，最后，再更新 sched.pidle，使其指向当前 p，这样，新的链表就构造完成。

接下来就要真正地准备休眠了，但是仍然不死心！还要再查看一次所有的 P 是否有工作，如果发现任何一个 P 有工作的话（判断 P 的本地队列不空），就先从全局空闲 P 链表里先拿到一个 P：

  1. // 试图从 _Pidle 列表里获取 p
  2. //go:nowritebarrierrec
  3. func pidleget() *p {
  4.     _p_ := sched.pidle.ptr()
  5.     if _p_ != nil {
  6.         sched.pidle = _p_.link
  7.         atomic.Xadd(&sched.npidle, -1) // TODO: fast atomic
  8.     }
  9.     return _p_
  10. }

比较简单，获取链表最后一个，再更新 sched.pidle，使其指向前一个 P。调用 acquirep(_p_) 绑定获取到的 p 和 m，主要的动作就是设置 p 的 m 字段，更改 p 的工作状态为 _Prunning，并且设置 m 的 p 字段。做完这些之后，再次进入 top 代码段，再走一遍之前找工作的过程。

  1. // 休眠，停止执行工作，直到有新的工作需要做为止
  2. func stopm() {
  3.     // 当前 goroutine，g0
  4.     _g_ := getg()
  6.     // ……………………
  7. retry:
  8.     lock(&sched.lock)
  9.     // 将 m 放到全局空闲链表里去
  10.     mput(_g_.m)
  11.     unlock(&sched.lock)
  12.     // 进入睡眠状态
  13.     notesleep(&_g_.m.park)
  14.     // 这里被其他工作线程唤醒
  15.     noteclear(&_g_.m.park)
  17.     // ……………………
  19.     acquirep(_g_.m.nextp.ptr())
  20.     _g_.m.nextp = 0
  21. }

先将 m 放入全局空闲链表里，注意涉及到全局变量的修改，要上锁。接着，调用 notesleep(&_g_.m.park) 使得当前工作线程进入休眠状态。其他工作线程在检测到“当前有很多工作要做”，会调用 noteclear(&_g_.m.park) 将其唤醒。注意，这两个函数传入的参数都是一样的：&_g_.m.park，它的类型是：

  1. type note struct {
  2.     key uintptr
  3. }

很简单，只有一个 key 字段。

note 的底层实现机制跟操作系统相关，不同系统使用不同的机制，比如 linux 下使用的 futex 系统调用，而 mac 下则是使用的 pthread_cond_t 条件变量，note 对这些底层机制做了一个抽象和封装。

这种封装给扩展性带来了很大的好处，比如当睡眠和唤醒功能需要支持新平台时，只需要在 note 层增加对特定平台的支持即可，不需要修改上层的任何代码。

上面这一段来自阿波张的系列教程。我们接着来看下 notesleep 的实现：

  1. // runtime/lock_futex.go
  2. func notesleep(n *note) {
  3.     // g0
  4.     gp := getg()
  5.     if gp != gp.m.g0 {
  6.         throw("notesleep not on g0")
  7.     }
  8.     // -1 表示无限期休眠
  9.     ns := int64(-1)
  11.     // ……………………
  13.     // 这里之所以需要用一个循环，是因为 futexsleep 有可能意外从睡眠中返回，
  14.     // 所以 futexsleep 函数返回后还需要检查 note.key 是否还是 0，
  15.     // 如果是 0 则表示并不是其它工作线程唤醒了我们，
  16.     // 只是 futexsleep 意外返回了，需要再次调用 futexsleep 进入睡眠
  17.     for atomic.Load(key32(&n.key)) == 0 {
  18.         // 表示 m 被阻塞
  19.         gp.m.blocked = true
  20.         futexsleep(key32(&n.key), 0, ns)
  22.         // ……………………
  24.         // 被唤醒，更新标志
  25.         gp.m.blocked = false
  26.     }
  27. }

继续往下追：

  1. // runtime/os_linux.go
  2. func futexsleep(addr *uint32, val uint32, ns int64) {
  3.     var ts timespec
  5.     if ns < 0 {
  6.         futex(unsafe.Pointer(addr), _FUTEX_WAIT, val, nil, nil, 0)
  7.         return
  8.     }
  10.     // ……………………
  11. }

当 *addr 和 val 相等的时候，休眠。futex 由汇编语言实现：

  1. TEXT runtime·futex(SB),NOSPLIT,$0
  2.     // 为系统调用准备参数
  3.     MOVQ    addr+0(FP), DI
  4.     MOVL    op+8(FP), SI
  5.     MOVL    val+12(FP), DX
  6.     MOVQ    ts+16(FP), R10
  7.     MOVQ    addr2+24(FP), R8
  8.     MOVL    val3+32(FP), R9
  9.     // 系统调用编号
  10.     MOVL    $202, AX
  11.     // 执行 futex 系统调用进入休眠，被唤醒后接着执行下一条 MOVL 指令
  12.     SYSCALL
  13.     // 保存系统调用的返回值
  14.     MOVL    AX, ret+40(FP)
  15.     RET

这样，找不到工作的 m 就休眠了。当其他线程发现有工作要做时，就会先找到空闲的 m，再通过 m.park 字段来唤醒本线程。唤醒之后，回到 findrunnable 函数，继续寻找 goroutine，找到后返回 schedule 函数，然后就会去运行找到的 goroutine。

这就是 m 找工作的整个过程，历尽千辛万苦，终于修成正果。

参考资料

【阿波张 Goroutine 调度策略】https://mp.weixin.qq.com/s/2objs5JrlnKnwFbF4a2z2g